package org.cloud.automatic.queue.job;

import lombok.extern.slf4j.Slf4j;
import org.cloud.automatic.queue.entity.SendQueueWorkerBO;
import java.util.HashMap;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;


@Slf4j
public class SendQueueMaster {
	
    /**
     * 任务队列
     */
    private ConcurrentLinkedQueue<SendQueueWorkerBO> workQueue = new ConcurrentLinkedQueue<SendQueueWorkerBO>();
    
    /**
     * 任务执行者
     */
    private HashMap<String, Thread> workers = new HashMap<String, Thread>();
    
    /**
     * 利用集合引用传递的特性，用来通知任务执行者终止任务。
     */
    private ConcurrentHashMap<String, String> noticeComplatedMap = new ConcurrentHashMap<String, String>();


    /**
     * 启动队列执行线程
     * @param worker 任务线程
     * @param num 启动线程数量
     */
    protected SendQueueMaster(Worker worker, int num){
        worker.setWorkQueue(workQueue);
        worker.setNoticeComplatedMap(noticeComplatedMap);
        for(int i = 0 ; i < num ; i++){
            Thread workerThread = new Thread(worker, "send-queue-thread-"+i);
            workers.put("send-queue-thread-"+i, workerThread);
        }
    }

    /**
     * 启动任务
     */
    public void executes(){
        Set<Entry<String, Thread>> entrys = workers.entrySet();
        for(Entry<String, Thread> thread : entrys){
            thread.getValue().start();
        }
    }
    
    /**
     * 提交一个任务
     * @param worker
     */
    public void submitTask(SendQueueWorkerBO worker){
        /**
         * 如果当前任务队列大于1000，则阻塞队列，直到资源被释放
         * 以免在补偿队列量巨大的时候造成内存OOM
         */
        while(workQueue.size() > maxQueueSize()) {
            try {
                Thread.currentThread().sleep(1);
            } catch (InterruptedException e) {
                log.error("====补偿队列阻塞等待异常====", e);
                Thread.currentThread().interrupt();
            }
        }
        
        workQueue.add(worker);
    }
    
    /**
     * 当前系统补偿队列允许最大值。
     * 可根据系统运行的环境调制此配置项。
     * 如果未配置或配置项小于等于0，则采用默认配置值1000.
     * 暂时默认1000
     */
    private int maxQueueSize() {
        return 1000 ;
    }

    
    public ConcurrentLinkedQueue<SendQueueWorkerBO> getWorkQueue() {
        return workQueue;
    }


    /**
     * 通知所有worker，如果获取不到新的任务，则结束任务
     */
    public void noticeCanStopWorker(){
        noticeComplatedMap.put("stop", "stop");
        //唤醒任务线程，通知终止任务
    }


}
